package com.bw.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @ProjectName: BigData
 * @Package: com.bw.app.dwd
 * @ClassName: BaseDBApp
 * @Author: Gy
 * @Description: 准备用户行为日志的DWD层
 * @Date: 2021/11/10 16:52
 */
public class BaseLogApp {

    private static final String TOPIC_START = "dwd_start_log";
    private static final String TOPIC_DISPLAY = "dwd_display_log";
    private static final String TOPIC_PAGE = "dwd_page_log";

    public static void main(String[] args) throws Exception {
        //todo 1.准备环境
        //1.1设置流式上下文环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2设置并行度
        env.setParallelism(1);

        //todo 2.从kafka中读取数据
        String topic = "ods_base_log";
        String groupId = "base_log_app_group";
        // 2.1 调用工具类，获取FinkKafkaConsumer
        FlinkKafkaConsumer<String> consumer = MyKafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> source = env.addSource(consumer);

        //todo 3.对读取到的数据进行转换    String->json
        SingleOutputStreamOperator<JSONObject> operator = source.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                JSONObject parse = JSON.parseObject(s);
                return parse;
            }
        });

        /*
        todo 4.识别新老客户
        */
        //4.1 根据mid对日志进行分组
        KeyedStream<JSONObject, String> midKey = operator.keyBy(
                data -> data.getJSONObject("common").getString("mid")
        );
        // 4.2 新老方法状态修复
        SingleOutputStreamOperator<JSONObject> jsonDSWithFlag = midKey.map(new RichMapFunction<JSONObject, JSONObject>() {
            //定义mid访问状态
            private ValueState<String> firstVisitDateState;
            //定义日期格式化对象
            private SimpleDateFormat sdf;

            @Override
            public void open(Configuration parameters) throws Exception {
                //对状态及日期格式进行初始化
                firstVisitDateState = getRuntimeContext().getState(
                        new ValueStateDescriptor<String>("newMidDateState", String.class)
                );
                sdf = new SimpleDateFormat("yyyyMMdd");
            }

            @Override
            public JSONObject map(JSONObject jsonObject) throws Exception {
                //获取当前日志标记状态
                String isNew = jsonObject.getJSONObject("common").getString("is_new");
                //获取当前日志访问时间戳
                Long ts = jsonObject.getLong("ts");

                if ("1".equals(isNew)) {
                    //获取当前mid对象的状态
                    String state = firstVisitDateState.value();
                    //对当前日志的日期格式进行转换
                    String format = sdf.format(new Date(ts));
                    //如果状态不为空，并且状态日期和当前日期不相等，说明是老客户
                    if (state != null && format.length() != 0) {
                        //判断是否为同一天数据
                        if (!state.equals(format)) {
                            isNew = "0";
                            jsonObject.getJSONObject("common").put("is_new", isNew);
                        }
                    } else {
                        //如果还没记录设备的状态，将当前访问日志作为状态值
                        firstVisitDateState.update(format);
                    }
                }
                return jsonObject;
            }
        });

        jsonDSWithFlag.print(">>>>>>>>>>");

        /*
        todo 5.分流
         */
        //定义启动测输出流标签
        OutputTag<String> startTag = new OutputTag<String>("start"){};
        //定义曝光测输出流标签
        OutputTag<String> displayTag = new OutputTag<String>("display"){};

        SingleOutputStreamOperator<String> pageDs = jsonDSWithFlag.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, Context context, Collector<String> collector) throws Exception {
                //获取启动日志标记
                JSONObject startJsonObj = jsonObject.getJSONObject("start");
                //将json格式转换为字符串，方便向测输出流输出以及向kafka写入
                String datastr = jsonObject.toString();

                //判断是否为启动日志
                if (startJsonObj != null && startJsonObj.size() > 0) {
                    //如果是启动日志，输出到启动测输出流
                    context.output(startTag, datastr);
                } else {
                    //如果不是启动日志，说明是页面日志，输出到主流
                    collector.collect(datastr);

                    //如果不是启动日志，获取曝光日志标记（曝光日志中也携带了页面）
                    JSONArray displays = jsonObject.getJSONArray("displays");
                    //判断是否为曝光日志
                    if (displays != null && displays.size() > 0) {
                        //如果是曝光日志，遍历输出到测输出流
                        for (int i = 0; i < displays.size(); i++) {
                            //获取每一条曝光事件
                            JSONObject displaysJSONObject = displays.getJSONObject(i);
                            //获取页面id
                            String pageId = jsonObject.getJSONObject("page").getString("page_id");
                            //给每一条曝光事件加pageId
                            displaysJSONObject.put("page_id", pageId);
                            context.output(displayTag, displaysJSONObject.toString());
                        }
                    }
                }
            }
        });

        //获取测输出流
        DataStream<String> startDs = pageDs.getSideOutput(startTag);
        DataStream<String> displayDs = pageDs.getSideOutput(displayTag);
        //打印输出
        pageDs.print("page>>>>>");
        startDs.print("start>>>>>");
        displayDs.print("display>>>>>");

        //todo 6.将不同流的数据写回到kafka的不同topic中
        FlinkKafkaProducer<String> startSink = MyKafkaUtil.getKafkaSink(TOPIC_START);
        startDs.addSink(startSink);

        FlinkKafkaProducer<String> displaySink = MyKafkaUtil.getKafkaSink(TOPIC_DISPLAY);
        displayDs.addSink(displaySink);

        FlinkKafkaProducer<String> pageSink = MyKafkaUtil.getKafkaSink(TOPIC_PAGE);
        pageDs.addSink(pageSink);

        env.execute();
   }
}
